package cn.itcast.hotel.mqListener;

import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.pojo.HotelDoc;
import cn.itcast.hotel.service.IHotelService;
import com.alibaba.fastjson.JSON;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;

@Component
public class MqListenerEsData {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private RestHighLevelClient client;

    @Resource
    IHotelService iHotelService;


    /**
     *  监听新增或者修改
     * @param msg
     * @throws InterruptedException
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "hotel.addOrUpdate"),
            exchange = @Exchange(name = "hotel.topic", type = ExchangeTypes.TOPIC),
            key = {"hotel.addOrUpdate"}
    ))
    public void listenTopicQueue1(String msg) throws InterruptedException {
        try {
            System.out.println("hotel.addOrUpdate接收到消息：" + msg);
            Long id = Long.valueOf(msg);
            Hotel hotel = iHotelService.getById(id);
            HotelDoc hotelDoc = new HotelDoc(hotel);
            IndexRequest indexRequest = new IndexRequest("hotel");
            indexRequest.timeout("1s");//设置超时时间
            indexRequest.id(hotel.getId().toString());//设置id
            indexRequest.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
            IndexResponse index = client.index(indexRequest, RequestOptions.DEFAULT);
            if (index.getResult().name().equals("CREATED")) {
                System.out.println("新增成功");
            } else if (index.getResult().name().equals("UPDATED")) {
                System.out.println("修改成功");
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("保存数据失败");
        }
    }


    /**
     *  删除
     * @param id
     * @throws InterruptedException
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "hotel.del"),
            exchange = @Exchange(name = "hotel.topic", type = ExchangeTypes.TOPIC),
            key = {"hotel.del"}
    ))
    public void listenTopicQueue2(String id) throws InterruptedException {
        try {
            DeleteRequest deleteRequest = new DeleteRequest();
            deleteRequest.id(id);
            deleteRequest.index("hotel");
            client.delete(deleteRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("删除失败");
        }
    }
}
